package io.github.eric.urpc.client;

import io.github.eric.urpc.client.handler.ClientConnectionHandler;
import io.github.eric.urpc.core.conn.Connection;
import io.github.eric.urpc.core.exception.UrpcConnectException;
import io.github.eric.urpc.core.exception.UrpcRuntimeException;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;

import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;

public class ClientConnector {
    private final Bootstrap bootstrap;
    private final ClientContext clientContext;

    ClientConnector(Bootstrap bootstrap, ClientContext clientContext) {
        this.bootstrap = bootstrap;
        this.clientContext = clientContext;
    }

    /**
     * 非阻塞式连接，直到成功为止
     */
    public void connectAsync() {
        InetSocketAddress remoteAddr = new InetSocketAddress(clientContext.getServerHost(), clientContext.getServerPort());
        clientContext.getLogger().info("[urpc client] start async connecting to urpc server {}", remoteAddr);
        connectAsync0(remoteAddr);
    }

    private void connectAsync0(InetSocketAddress remoteAddr) {
        try {
            // 如果是主动shutdown则不进行重连
            if (clientContext.isShuttingDown()) {
                clientContext.getLogger().info("[urpc client] client is shutdown now, drop async connect task to addr {}", remoteAddr);
                return;
            }
            bootstrap.connect(remoteAddr).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (clientContext.isShuttingDown()) {
                        return;
                    }
                    Channel channel = future.channel();
                    if (!channel.isActive()) {
                        if (channel.isRegistered()) {
                            channel.close();
                        }
                        try {
                            clientContext.getHashedWheelTimer().newTimeout(new TimerTask() {
                                public void run(Timeout timeout) throws Exception {
                                    connectAsync0(remoteAddr);
                                }
                            }, 1, TimeUnit.SECONDS);
                        } catch (Exception e) {
                            clientContext.getLogger().error("[urpc client] create async connect task to addr {} failed!", remoteAddr, e);
                        }
                    }
                }
            });
        } catch (Exception e) {
            clientContext.getLogger().error("[urpc client] async connecting to {} error!", remoteAddr, e);
        }
    }

    /**
     * 阻塞式连接，直到成功为止
     */
    public Connection connectSync() {
        try {
            if (clientContext.isShuttingDown()) {
                throw new UrpcRuntimeException("[urpc client] client is shutting down, stop connect sync");
            }
            return connectOnce();
        } catch (UrpcRuntimeException ue) {
            throw ue;
        } catch (Exception e) {
            try {
                Thread.sleep(1000L);
            } catch (InterruptedException ex) {
                Thread.currentThread().interrupt();
                throw new UrpcRuntimeException("[urpc client] connect sync interrupted, cause: " + ex.getMessage());
            }
            return connectSync();
        }
    }

    /**
     * 只进行一次连接
     */
    public Connection connectOnce() {
        ChannelFuture f = null;
        try {
            InetSocketAddress remoteAddr = new InetSocketAddress(clientContext.getServerHost(), clientContext.getServerPort());
            clientContext.getLogger().info("[urpc client] start sync connecting to urpc server {}", remoteAddr);
            f = bootstrap.connect(remoteAddr).syncUninterruptibly();
            f.channel().pipeline().get(ClientConnectionHandler.class).getChannelActiveLatch().await();
            if (f.channel().isActive()) {
                clientContext.getLogger().info("[urpc client] connect succeed, remote addr {}", remoteAddr);
                return f.channel().attr(Connection.TCP).get();
            } else {
                throw f.cause();
            }
        } catch (Throwable e) {
            if (f != null) {
                f.channel().close();
            }
            throw new UrpcConnectException(String.format("[urpc client] connect to urpc server failed, cause: %s", e.getMessage(), e));
        }
    }

}
